#include "EpollAsynchIO.h"
#include "MessageBlock.h"
#include "EpollProactor.h"
#include "Debug.h"
#include "Trace.h"

EpollAsynchOpt::EpollAsynchOpt (EpollProactor *ep)
: epoll_proactor_ (ep)
, proactor_ (0)
, asynch_handler_ (0)			      
, handle_ (NDK_INVALID_HANDLE)
, result_ (0)
{
    TRACE ("");
}
EpollAsynchOpt::~EpollAsynchOpt ()
{
    TRACE ("");
    if (this->result_)
    {
        delete this->result_;
        this->result_ = 0;
    }
}
int EpollAsynchOpt::open (const AsynchHandler *handler,
        NDK_HANDLE handle,
        const void * /* completion_key */,
        Proactor *proactor)
{
    TRACE ("");
    if (handler == 0) return -1;

    this->proactor_ = proactor;
    this->asynch_handler_ = const_cast<AsynchHandler *>(handler);
    this->handle_ = handle;

    if (this->handle_ == NDK_INVALID_HANDLE)
        this->handle_ = handler->handle ();

    if (this->handle_ == NDK_INVALID_HANDLE)
        return -1;

    return 0;
}
// ---------------------------------------------------------------------
EpollAsynchResult::EpollAsynchResult (const AsynchHandler *handler,
	NDK_HANDLE handle,
	long long offset)
: handle_ (handle)
, bytes_transferred_ (0)
, success_ (0)
, completion_key_ (0)
, asynch_handler_ (const_cast<AsynchHandler*>(handler))
, error_ (0)
, offset_ (offset)
{
    TRACE ("");
}
EpollAsynchResult::~EpollAsynchResult ()
{
    TRACE ("");
}
// ---------------------------------------------------------------------
EpollAsynchReadStream::EpollAsynchReadStream (EpollProactor *p)
: AsynchReadStreamImpl () 
, EpollAsynchOpt (p)
{
    TRACE ("");
}
EpollAsynchReadStream::~EpollAsynchReadStream ()
{
    TRACE ("");
}
int EpollAsynchReadStream::open (const AsynchHandler *handler,
        NDK_HANDLE handle,
        const void *completion_key,
        Proactor *proactor/* = 0*/)
{
    TRACE ("");
    if (EpollAsynchOpt::open (handler,
                handle, 
                completion_key,
                proactor) == 0)
    {
        this->result_ = new EpollAsynchReadStreamResult(
                this->asynch_handler_,
                this->handle_,
                0,
                0);
        return this->epoll_proactor_->register_handle (this->handle_,
                0,
                AsynchHandler::READ_MASK);
    }
    return -1;
}
int EpollAsynchReadStream::cancel (void)
{
    TRACE ("");
    if (!this->epoll_proactor_)
        return -1;

    return this->epoll_proactor_->cancel_aio (this->handle_,
            AsynchHandler::READ_MASK);
}
int EpollAsynchReadStream::read (MessageBlock *message_block,
        size_t bytes_to_read)
{
    TRACE ("");
    if (message_block == 0) return -1;
    if (bytes_to_read > message_block->size ())
        bytes_to_read = message_block->size ();
    if (bytes_to_read == 0)
        return -1;
    if (this->result_ == 0) return -1;
    /*
       EpollAsynchReadStreamResult *result = new EpollAsynchReadStreamResult(
       this->asynch_handler_,
       this->handle_,
       message_block,
       bytes_to_read);
       */
    EpollAsynchReadStreamResult *r 
        = dynamic_cast<EpollAsynchReadStreamResult *>(this->result_);
    r->message_block (message_block);
    r->bytes_to_read (bytes_to_read);
    r->bytes_to_read_back_ = bytes_to_read;
    return this->proactor_impl ()->start_read (this); 
}
int EpollAsynchReadStream::execute (EpollAsynchResult *&r_result)
{
    TRACE ("");
    int result = 0;
    int finished = 0;

    r_result = this->result_;
    EpollAsynchReadStreamResult *read_result = 
        dynamic_cast<EpollAsynchReadStreamResult *>(this->result_);
    read_result->success_ = 1;
    read_result->error_ = 0;   // refresh;
    read_result->bytes_transferred_ = 0;
    // Not loop
    result = NDK::recv (this->handle_,
            read_result->message_block_->wr_ptr (), 
            read_result->bytes_to_read_back_,
            0);
    if (result > 0) 
    {
        if (static_cast<size_t>(result) 
                < read_result->bytes_to_read_back_)  
        {
            finished = 1;     // special for Level Triggered model.
            //finished = 0;   // special for Edge Triggered model.
        }else
        {
            finished = 1; // read complete
        }
        read_result->message_block_->wr_ptr (result);
        read_result->bytes_transferred_ += result;
        read_result->bytes_to_read_back_ -= result;
    }else if (result < 0)
    {
        if (errno == EAGAIN)  // special for Edge Triggered model.
        {
            read_result->error_ = EAGAIN;
            read_result->success_ = 0;
        }else  // error
        {
            read_result->error_ = errno;
            read_result->success_ = 0;
        }
        finished = 1;
    }else  // result == 0
        finished = 1;

    //read_result->bytes_transferred_ = result == -1 ? 0 : result;
    return finished;
}
EpollAsynchReadStreamResult::EpollAsynchReadStreamResult (
        const AsynchHandler *handler,
        NDK_HANDLE handle,
        MessageBlock *message_block,
        size_t bytes_to_read)
: AsynchReadStreamResultImpl ()
, EpollAsynchResult (handler, handle, 0)
, bytes_to_read_ (bytes_to_read)
, bytes_to_read_back_ (bytes_to_read)
, message_block_ (message_block)
{
    TRACE ("");
}
EpollAsynchReadStreamResult::~EpollAsynchReadStreamResult ()
{
    TRACE ("");
}
// ---------------------------------------------------------------------
EpollAsynchWriteStream::EpollAsynchWriteStream (EpollProactor *p)
: AsynchWriteStreamImpl () 
, EpollAsynchOpt (p)
{
    TRACE ("");
}
EpollAsynchWriteStream::~EpollAsynchWriteStream ()
{
    TRACE ("");
}
int EpollAsynchWriteStream::open (const AsynchHandler *handler,
        NDK_HANDLE handle,
        const void * completion_key,
        Proactor *proactor)
{
    TRACE ("");
    if (EpollAsynchOpt::open (handler,
                handle,
                completion_key,
                proactor) == 0)
    {
        this->result_ = new EpollAsynchWriteStreamResult(
                this->asynch_handler_,
                this->handle_,
                0,
                0);
        return 0;
    }
    return -1;
}
int EpollAsynchWriteStream::cancel (void)
{
    TRACE ("");
    if (!this->epoll_proactor_)
        return -1;

    return this->epoll_proactor_->cancel_aio (this->handle_,
            AsynchHandler::WRITE_MASK);
}
int EpollAsynchWriteStream::write (MessageBlock *message_block,
        size_t bytes_to_write)
{
    TRACE ("");
    if (message_block == 0) return -1;
    if (bytes_to_write > message_block->length ())
        bytes_to_write = message_block->length ();
    if (bytes_to_write == 0)
        return -1;
    EpollAsynchWriteStreamResult *result = new EpollAsynchWriteStreamResult(
            this->asynch_handler_,
            this->handle_,
            message_block,
            bytes_to_write);
    //if (this->result_ == 0) return -1;
    return this->proactor_impl ()->start_write (this, result); 
}
int EpollAsynchWriteStream::execute (EpollAsynchResult *&w_result)
{
    TRACE ("");
    int result = 0;
    int finished = 0;

    EpollAsynchWriteStreamResult *write_result = 
        dynamic_cast<EpollAsynchWriteStreamResult *>(w_result);
    write_result->success_ = 1;
    write_result->bytes_transferred_ = 0;
    write_result->error_ = 0;   // refresh;
    result = NDK::send (this->handle_,
            write_result->message_block_->rd_ptr (), 
            write_result->bytes_to_write_back_,
            0);
    if (result > 0) 
    {
        if (static_cast<size_t>(result) 
                < write_result->bytes_to_write_back_)
        {
            //finished = 1;   
            finished = 0; 
        }else
        {
            finished = 1;
        }
        write_result->message_block_->rd_ptr (result);
        write_result->bytes_transferred_ += result;
        write_result->bytes_to_write_back_ -= result;
    }else if (result < 0)
    {
        if (errno == EAGAIN)  // special for Edge Triggered model.
        {
            write_result->error_ = EAGAIN;
            finished = 0;
        }else  // error
        {
            write_result->error_ = errno;
            write_result->success_ = 0;
            finished = 1;
        }
    }else  // result == 0
        finished = 1;

    //write_result->bytes_transferred_ = result == -1 ? 0 : result;
    return finished;
}
EpollAsynchWriteStreamResult::EpollAsynchWriteStreamResult (
        const AsynchHandler *handler,
        NDK_HANDLE handle,
        MessageBlock *message_block,
        size_t bytes_to_write)
: AsynchWriteStreamResultImpl ()
, EpollAsynchResult (handler, handle, 0)
, bytes_to_write_ (bytes_to_write)
, bytes_to_write_back_ (bytes_to_write)
, message_block_ (message_block)
{
    TRACE ("");
}
EpollAsynchWriteStreamResult::~EpollAsynchWriteStreamResult ()
{
}
// ---------------------------------------------------------------------
EpollAsynchAccept::EpollAsynchAccept (EpollProactor *p)
: AsynchAcceptImpl ()
, EpollAsynchOpt (p)
{
    TRACE ("");
}
EpollAsynchAccept::~EpollAsynchAccept ()
{
    TRACE ("");
}
int EpollAsynchAccept::open (const AsynchHandler *handler,
        NDK_HANDLE handle,
        const void *completion_key,
        Proactor *proactor)
{
    TRACE ("");
    if (EpollAsynchOpt::open (handler,
                handle,
                completion_key,
                proactor) == 0)
    {
        this->result_ = new EpollAsynchAcceptResult(
                this->asynch_handler_,
                this->handle_,
                0,
                0);
        return this->epoll_proactor_->register_handle (this->handle_,
                0,
                AsynchHandler::READ_MASK);
    }
    return -1;
}
int EpollAsynchAccept::cancel (void)
{
    TRACE ("");
    if (!this->epoll_proactor_)
        return -1;

    return this->epoll_proactor_->cancel_aio (this->handle_,
            AsynchHandler::READ_MASK);
}
int EpollAsynchAccept::accept (MessageBlock *message_block,
        size_t num_to_accept)
{
    TRACE ("");
    if (message_block == 0) return -1;
    if (sizeof (ConnectionInfo) * num_to_accept > message_block->size ())
    {
        num_to_accept = message_block->size () / sizeof (ConnectionInfo);
    }
    if (num_to_accept == 0)
        return -1;
    if (this->result_ == 0) return -1;

    /*
       EpollAsynchAcceptResult *result = new EpollAsynchAcceptResult(
       this->asynch_handler_,
       message_block,
       this->handle_,
       num_to_accept);
       */
    EpollAsynchAcceptResult *r = 
        dynamic_cast<EpollAsynchAcceptResult *>(this->result_);
    r->message_block (message_block);
    r->num_to_accept (num_to_accept);
    return this->proactor_impl ()->start_read (this); 
}
int EpollAsynchAccept::execute (EpollAsynchResult *&apt_result)
{
    TRACE ("");
    int finished = 1;
    size_t num_accepted = 0;
    NDK_HANDLE accept_handle = NDK_INVALID_HANDLE;

    apt_result = this->result_;
    EpollAsynchAcceptResult *accept_result =
        dynamic_cast<EpollAsynchAcceptResult *>(this->result_);
    accept_result->success_ = 1;
    accept_result->error_ = 0;   // refresh;

    while(num_accepted < accept_result->num_to_accept_)
    {
        InetAddr remote_addr;
        sockaddr *addr   = 0;
        addr    = (sockaddr *) remote_addr.get_addr ();
        int len = remote_addr.get_addr_size ();
        accept_handle = NDK_OS::accept (accept_result->listen_handle_,
                addr,
                &len);
        if (accept_handle == -1)
        {
            if (errno == EAGAIN)  // special for Edge Triggered model.
            {
                accept_result->error_ = EAGAIN;
                break;
            }else if (errno == EINTR)
            {
                continue;
            }else  // error
            {
                accept_result->error_ = errno;
                accept_result->success_ = 0;
                break;
            }
        }
        //NDK_INF ("recv a connection %d, listen handle = %d\n", 
        //accept_handle,
        //accept_result->listen_handle_);
        ConnectionInfo info (remote_addr, accept_handle);
        ::memcpy (accept_result->message_block_->wr_ptr (), 
                &info, 
                sizeof (info));
        accept_result->message_block_->wr_ptr (sizeof (info));
        ++num_accepted;
    }
    return finished;
}
EpollAsynchAcceptResult::EpollAsynchAcceptResult (
        const AsynchHandler *handler,
        NDK_HANDLE listen_handle,
        MessageBlock *message_block,
        size_t num_to_accept)
: AsynchAcceptResultImpl ()
, EpollAsynchResult (handler, listen_handle, 0)
, listen_handle_ (listen_handle)
, num_to_accept_ (num_to_accept)
, message_block_ (message_block)
{
    TRACE ("");
}
EpollAsynchAcceptResult::~EpollAsynchAcceptResult ()
{
    TRACE ("");
}

